/*
 * proxy_application.cpp
 *
 *  Created on: 2013-12-19
 *      Author: lixingyi
 */

#include <string>
#include "framework/system_util.h"
#include "third_party/pugixml.hpp"

#include "proxy_application.h"
#include "version.h"
#include "proxy_thread.h"
#include "mysql_thread.h"
#include "log_thread.h"

using namespace std;
using namespace kingnet ;

enum { LARGE_BUF_SIZE = 8388608 };

ClientAcceptor::ClientAcceptor()
{

}

int ClientAcceptor::on_connection(int fd,sa_in_t* addr)
{
    //std::shared_ptr<ProxyThread> proxy_thread ;
    
    /*
    shared_ptr<ProxyThread> thread(new ProxyThread) ;
    if(thread.get() == NULL ) return -1 ;
    proxy_thread_container[fd] = thread ;
    */
    shared_ptr<ProxyThread>& thread =proxy_thread_container[fd]  ;
    thread.reset(new ProxyThread) ;
    if(thread.get() == NULL ) return -1 ;
    debug_log_format(get_app().logger,"new_connection , fd:%d\n",fd) ;
    thread->set_args(fd) ;
    thread->start();
    thread->detach() ;

    return 0 ;
}

int MysqlThreadManager::init(const char* config_file)
{
    pugi::xml_document xml_config ;
    if(!xml_config.load_file(config_file))
    {
        error_return(-1,"load config failed") ;
    }
    pugi::xml_node root = xml_config.child("root") ;
    if(!root) error_return(-1,"missing <root> node") ;

    bool thread_failed = false ;
    pugi::xml_node node = root.child("mysql_server_list");
    for (pugi::xml_node input = node.first_child(); input;
        input = input.next_sibling())
    {
        MysqlConfig mysql_data ;
        mysql_data.reset( input.attribute("host").value(),
            input.attribute("username").value(),
            input.attribute("password").value(),
            input.attribute("dbname").value(),
            input.attribute("charset").value(),
            input.attribute("id").as_int(),
            input.attribute("port").as_int() ) ;

        auto& thread = m_thread_container[mysql_data.id()] ;
        thread.reset(new MysqlThread) ;
        if(thread.get() == NULL )
        {
            thread_failed = true ;
            break ;
        }

        thread->set_args(mysql_data) ;
        thread->start() ;
        
    }



    if(thread_failed) error_return(-1,"create thread failed") ;

    for(int i = 0 ; i < 3 ; ++i)
    {
        if(!is_all_running()) sleep(1);
    }
    if(!is_all_running()) error_return(-1,"create thread failed") ;

    m_default_thread = get_mysql_thread(0) ;
    if(m_default_thread == NULL ) error_return(-1,"no default mysql_thread") ;

    return 0 ;

}

bool MysqlThreadManager::is_all_running()
{
    for(auto& value : m_thread_container)
    {
        if(value.second.get()==NULL || (!value.second->running()) ) return false ;
        
    }

    return true ;

}

void MysqlThreadManager::fini()
{
    for(auto & value : m_thread_container)
    {
        if(value.second.get())
        {
            //info_log_format(logger,"stop mysql_thread:%ld\n",value.second->id()) ;
            value.second->stop() ;
            value.second->join() ;
        }
    }



}


MysqlThread* MysqlThreadManager::get_mysql_thread(int id)
{
    auto it = m_thread_container.find(id) ;
    if(it != m_thread_container.end() ) return it->second.get() ;
    return m_default_thread ;
}


ProxyApplication::ProxyApplication()
{
    m_is_shuting_down = false ;

}

ProxyApplication::~ProxyApplication()
{
    // TODO Auto-generated destructor stub
}

const char* ProxyApplication::version()
{
    return VERSION ;
}

int ProxyApplication::on_init()
{

    pugi::xml_document xml_config ;
    if(!xml_config.load_file(config_file))
    {
        error_return(-1,"load config failed") ;
    }
    pugi::xml_node root = xml_config.child("root") ;
    if(!root) error_return(-1,"missing <root> node") ;

    pugi::xml_node node = root.child("log") ;
    if(logger.init(node.attribute("prefix").value(),
            node.attribute("level").as_int() ) !=0 )
    {
        error_return(-1,"init logger failed") ;
    }

    if(m_log_thread.get()==NULL) m_log_thread.reset(new LogThread) ;
    if(m_log_thread.get() == NULL ) error_return(-1,"create log thread failed") ;
    m_log_thread->set_args(logger.get_prefix(),logger.get_level()) ;
    m_log_thread->start() ;
    usleep(1000) ;

    node = root.child("client_acceptor") ;
    if(m_client_acceptor.init(&event_engine,
            node.attribute("listen_ip").value(),
            node.attribute("listen_port").as_int() )!=0)
    {
        error_return(-1,"init client acceptor failed") ;
    }
    //set_defer_accept(m_client_acceptor.fd(),10 ) ;

    MysqlThreadManager* mysql_container = new MysqlThreadManager;
    if( mysql_container == NULL ) error_return(-1,"create mysql_thread failed");
    if( mysql_container->init(config_file)!=0 )
    { 
        delete mysql_container ;
        error_return(-1,"create mysql_thread failed");
    }

    //prevent reorder
    MysqlThreadManager* old_manager = m_mysql_manager.load() ;
    m_mysql_manager.store(mysql_container,memory_order_acquire) ;
    if(old_manager)
    { 
        m_delayed_manager[time_ms()+10000] = old_manager ;
    }

    

   
    set_timer(60000);

    //set_delay_stop(5000);

    srand(time(0));
    return 0 ;
}

int ProxyApplication::on_reload()
{
    m_client_acceptor.fini() ;

    if(on_init()!=0) 
    {
        info_log_format(logger,"reload failed") ;       
    }
    else
    {
        info_log_format(logger,"reload success") ;       
    }

    return 0 ;
}

void ProxyApplication::on_delay_stop()
{
    if (m_is_shuting_down)
    {
        return;
    }
    m_is_shuting_down = true;

    m_client_acceptor.fini() ;
}

void ProxyApplication::on_fini()
{

    m_client_acceptor.fini() ;
    
    for(auto & value : m_client_acceptor.proxy_thread_container)
    {
        if(value.second.get())
        {
            info_log_format(logger,"stop proxy_thread:%ld\n",value.second->id()) ;
            value.second->stop() ;
        }
    }
    //m_client_acceptor.proxy_thread_container.clear() ;

    MysqlThreadManager* mysql_manager = m_mysql_manager.load() ;
    if(mysql_manager ) delete m_mysql_manager ;
    for(auto & value : m_delayed_manager) delete value.second ;

    m_log_thread->stop() ;
    m_log_thread->join() ;

    info_log_string(logger,"system stopped") ;
}


MysqlThread* ProxyApplication::get_mysql_thread(int id)
{
    return m_mysql_manager.load()->get_mysql_thread(id) ;
}

void ProxyApplication::on_timer()
{
    //free old mysql manager
    auto it = m_delayed_manager.begin() ;
    if(it!=m_delayed_manager.end() && time_ms() > it->first)
    {
        delete it->second ;
        m_delayed_manager.erase(it) ;
        info_log_string(logger,"destroy old mysql manager") ;
    }

}


IMPLEMENT_APPLICATION_INSTANCE(ProxyApplication) ;
IMPLEMENT_MAIN() ;
